package POE::Component::Gearman::Worker;

use 5.006;
use strict;
use warnings FATAL => 'all';

use Smart::Comments;

use POE qw(
    Component::Gearman::Connection
    Component::Gearman::Job
);

=head1 NAME

POE::Component::Gearman::Worker - Asynchronous Gearman worker for POE applications 

=head1 VERSION

Version 0.01

=cut

our $VERSION = '0.01';

=head1 SYNOPSIS

    use POE qw(Component::Gearman);

    spawn POE::Component::Gearman {
        Alias       => 'Gearman',
        JobServers  => [ '127.0.0.1', '192.168.0.1:123' ],
    };

    run POE::Kernel;

=head1 EXPORT

A list of functions that can be exported.  You can delete this section
if you don't export anything, such as for a purely object-oriented module.

=head1 SUBROUTINES/METHODS

=head2 spawn

=cut

sub spawn {
    shift if defined $_[0] and $_[0] eq __PACKAGE__;
    my $cfg = shift || {};

    create POE::Session(
        package_states => [
            (__PACKAGE__) => [qw(
                _start
                register on_connection_ready
                _on_job_changed
                on_NOOP on_NO_JOB on_JOB_ASSIGN on_JOB_ASSIGN_UNIQ
            )],
            'POE::Component::Gearman::Connection' => [qw(
                on_tcp_connected on_connect_error reconnect
                on_read_write_error
                on_recv_command send_command
            )],
        ],
        heap => {
            box => {},
            cfg => $cfg,
        },
    );

    return;
}

sub _start {
    my ($kernel, $heap) = @_[ KERNEL, HEAP ];
    my $cfg = $heap->{cfg};

    $cfg->{Alias}       ||= 'GearmanWorker';
    $cfg->{JobServers}  ||= [ qw(127.0.0.1) ];
    $cfg->{Prefix}      ||= '';

    $kernel->alias_set( $cfg->{Alias} );

    foreach my $connStr ( @{ $cfg->{JobServers} } ){
        1 while $connStr =~ s/^(\s+)|(\s+)$//g;

        my ($host, $port);
        if ( $connStr =~ /^(.*):(\d+)$/ ){
            $host = $1;
            $port = $2;
        }
        else{
            $host = $connStr;
            $port = 4730;
        }

        my $wheel = new POE::Component::Gearman::Connection( $host, $port );
        my $id = $wheel->ID;

        $heap->{connections}{$id} = {
            connStr     => $connStr,
            state       => 'connect',
            host        => $host,
            port        => $port,
            sock_wheel  => $wheel,
        };
    }
}

sub register {
    my ($kernel, $heap) = @_[ KERNEL, HEAP ];
    my $sender = $_[ SENDER ];
    my @funcs = @_[ ARG0..$#_ ];

    my $cfg = $heap->{cfg};

    foreach my $f ( @funcs ){
        my $funcName = $f;
        if ( $cfg->{Prefix} ne '' ){
            $funcName = $cfg->{Prefix} . $f;
        }

        $heap->{functions}{ $funcName } = {
            sender  => $sender,
            func    => $f,
        };

        foreach my $connDesc ( values %{ $heap->{connections} } ){
            next if $connDesc->{state} ne 'ready';
            my $connID = $connDesc->{rw_wheel}->ID;
            $kernel->yield( send_command => $connID => CAN_DO => $funcName );
        }
    }
}

sub on_connection_ready {
    my ($kernel, $heap) = @_[ KERNEL, HEAP ];
    my $connID = $_[ ARG0 ];

    foreach my $f ( keys %{ $heap->{functions} } ){
        $kernel->yield( send_command => $connID => CAN_DO => $f );
    }

    $heap->{connections}{$connID}->{state} = 'ready';

    $kernel->yield( send_command => $connID => 'GRAB_JOB_UNIQ' );
}

sub on_NOOP {
    my ($kernel, $heap) = @_[ KERNEL, HEAP ];
    my $connID = $_[ ARG1 ];

    $kernel->yield( send_command => $connID => 'GRAB_JOB_UNIQ' );
}

sub on_NO_JOB {
    my ($kernel, $heap) = @_[ KERNEL, HEAP ];
    my $connID = $_[ ARG1 ];

    $kernel->yield( send_command => $connID => 'PRE_SLEEP' );
}

sub on_JOB_ASSIGN {
    my ($kernel, $session, $heap) = @_[ KERNEL, SESSION, HEAP ];
    my ($jobHandle, $function, $data) = @{ $_[ARG0] };
    my $connID = $_[ ARG1 ];

    my $handler = $heap->{functions}{$function};
    if ( not defined $handler ){
        # TODO:
        $kernel->yield( send_command => $connID => 'PRE_SLEEP' );
        return;
    }

    my $job = new POE::Component::Gearman::Job(
        handle      => $jobHandle,
        function    => $function,
        workload    => $data,
        callback    => $session->postback( _on_job_changed => $jobHandle, $connID ),
    );

    $kernel->post( $handler->{sender}, $handler->{func}, $job );
}

sub on_JOB_ASSIGN_UNIQ {
    my ($kernel, $session, $heap) = @_[ KERNEL, SESSION, HEAP ];
    my ($jobHandle, $function, $uniqID, $data) = @{ $_[ARG0] };
    my $connID = $_[ ARG1 ];

    my $handler = $heap->{functions}{$function};
    if ( not defined $handler ){
        # TODO:
        $kernel->yield( send_command => $connID => 'PRE_SLEEP' );
        return;
    }

    my $job = new POE::Component::Gearman::Job(
        handle      => $jobHandle,
        function    => $function,
        workload    => $data,
        callback    => $session->postback( _on_job_changed => $jobHandle, $connID ),
    );

    $kernel->post( $handler->{sender}, $handler->{func}, $job );
}

sub _on_job_changed {
    my ($kernel, $heap)         = @_[ KERNEL, HEAP ];
    my ($cbArgs, $args)         = @_[ARG0..$#_];
    my ($jobHandle, $connID)    = @$cbArgs;
    my ($command, @data)        = @$args;

    $kernel->yield( send_command => $connID => $command => $jobHandle => @data );

    if ( $command eq 'WORK_COMPLETE' ){
        $kernel->yield( send_command => $connID => 'GRAB_JOB_UNIQ' );
    }
}

1;

__END__
todo list:

# Worker 的停止、暂停、继续
# Worker 增加连接、减少连接
